package com.scott;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author 黄后林
 * @version 1.0
 * @note  使用flink做批处理  数据是到齐的 是现成的
 * @date 2022-01-22 00:08:43
 */
public class WordCount {

    public static void main(String[] args) throws Exception {

        // 1、 创建执行环境 = 上下文
        ExecutionEnvironment evn = ExecutionEnvironment.getExecutionEnvironment();

        // 2、 从文件中读取数据
        String file ="src/main/resources/hello.txt";

        // 3、 读取文本文件从而得到1个数据源   算子 = 运算符 = 数据集
        DataSet<String> inputDataSet = evn.readTextFile(file);

        // 4、 对数据集进行处理
        // 先用空格拆开、拆成1个1个的word 并转化成(word, 1) 二元组

        DataSet<Tuple2<String,Integer>> ds = inputDataSet
                                                    // 先用空格拆开、拆成1个1个的word
                                                    .flatMap( new ScottFlatMapFunction())
                                                    // 安装第1个位置的word进行分组
                                                    .groupBy(0)
                                                    // 把第2个位置的数字相加
                                                    .sum(1);
        ds.print();
    }

    public static class ScottFlatMapFunction implements FlatMapFunction<String, Tuple2<String,Integer>> {

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {

            // 1、 按空格进行分词
            String[] words = value.split(" ");

            // 2、 遍历所有word 封装成二元组
            for(String word: words){
                Tuple2<String, Integer> t2 = new Tuple2<>(word, 1);
                out.collect(t2);
            }

        }
    }
}
